package com.xrui.hbase.impl;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;

import com.xrui.hbase.BufferedWriter;
import com.xrui.hbase.ColumnName;
import com.xrui.hbase.EntityId;
import com.xrui.hbase.HBaseTable;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.util.DebugResourceTracker;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;


/**
 * <p>
 * HBase implementation of a batch TableWriter.  Contains its own HTable connection to optimize
 * performance.  Buffer is stored locally and the underlying HTableInterface buffer is ignored.
 * Default buffer size is 2,097,152 bytes.
 * </p>
 * <p>
 * <p>
 * Access to this Writer is threadsafe.  All internal state mutations must synchronize against
 * mInternalLock.
 * </p>
 */
public final class HBaseBufferedWriter implements BufferedWriter {
    //region static
    private static final Logger LOG = LoggerFactory.getLogger(HBaseBufferedWriter.class);
    //endregion

    //region variables
    /**
     * Underlying dedicated HTableInterface used by this writer. Owned by this writer.
     */
    private final HTableInterface mHTable;

    /**
     * Table this writer is attached to.
     */
    private final HBaseTable mTable;

    /**
     * Layout consumer registration resource.
     */
    private final TableLayout mTableLayout;

    /**
     * Monitor against which all internal state mutations must be synchronized.
     */
    private final Object mInternalLock = new Object();
    /**
     * Static overhead size of a Delete.
     */
    private final long mDeleteSize = ClassSize.align(
        ClassSize.OBJECT + 2 * ClassSize.REFERENCE
            + 2 * Bytes.SIZEOF_LONG + Bytes.SIZEOF_BOOLEAN
            + ClassSize.REFERENCE + ClassSize.TREEMAP);
    /**
     * Local write buffers.
     */
    private Map<EntityId, Put> mPutBuffer = new HashMap<EntityId, Put>();
    private ArrayList<Delete> mDeleteBuffer = Lists.newArrayList();
    /**
     * Local write buffer size.
     */
    private long mMaxWriteBufferSize = 1024L * 1024L * 2L;
    private long mCurrentWriteBufferSize = 0L;
    /**
     * Tracks the state of this buffered writer.
     * Reads and writes to mState must by synchronized by mInternalLock.
     */
    private State mState = State.UNINITIALIZED;

    /**
     * Creates a buffered table writer that stores modifications to be sent on command
     * or when the buffer overflows.
     *
     * @param table A table.
     * @throws IOException in case of IO errors.
     */
    public HBaseBufferedWriter(HBaseTable table) throws IOException {
        mTable = table;

        mHTable = mTable.openHTableConnection();

        mTableLayout = mTable.getLayout();
        mHTable.setAutoFlush(false);

        // Retain the table only after everything else succeeded:
        mTable.retain();
        synchronized (mInternalLock) {
            Preconditions.checkState(mState == State.UNINITIALIZED,
                "Cannot open HBaseBufferedWriter instance in state %s.", mState);
            mState = State.OPEN;
        }
        DebugResourceTracker.get().registerResource(this);
    }

    //endregion

    //region construct

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> void put(EntityId entityId, String family, String qualifier, T value)
        throws IOException {
        put(entityId, family, qualifier, HConstants.LATEST_TIMESTAMP, value);
    }
    //endregion

    //regionAPIs
    //region put

    /**
     * {@inheritDoc}
     */
    @Override
    public <T> void put(EntityId entityId, String family, String qualifier, long timestamp, T value)
        throws IOException {
        //TODO: if code decoded is implemented, we should use columnName instead of using the family and qualifier directly.
        //final ColumnName columnName = ColumnName.create(family, qualifier);
        //TODO: wait for encode implement

        updateBuffer(entityId, family.getBytes(), qualifier.getBytes(), timestamp, "To Be implements".getBytes());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteRow(EntityId entityId) throws IOException {
        deleteRow(entityId, HConstants.LATEST_TIMESTAMP);
    }
    //endregion

    //region delete

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteRow(EntityId entityId, long upToTimestamp) throws IOException {
        final Delete delete = new Delete(entityId.getHBaseRowKey(), upToTimestamp);
        updateBuffer(delete);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteFamily(EntityId entityId, String family) throws IOException {
        deleteFamily(entityId, family, HConstants.LATEST_TIMESTAMP);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteFamily(EntityId entityId, String family, long upToTimestamp)
        throws IOException {
        // The only data in this ETS and DataCloud HBase family is the one cf family, so we can delete everything.
        final ColumnName columnName = ColumnName.create(family);
        final Delete delete = new Delete(entityId.getHBaseRowKey());
        delete.deleteFamily(columnName.getFamilyBytes(), upToTimestamp);

        // Buffer the delete.
        updateBuffer(delete);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteColumn(EntityId entityId, String family, String qualifier) throws IOException {
        deleteColumn(entityId, family, qualifier, HConstants.LATEST_TIMESTAMP);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteColumn(EntityId entityId, String family, String qualifier, long upToTimestamp)
        throws IOException {
        final ColumnName hbaseColumnName = ColumnName.create(family, qualifier);
        final Delete delete = new Delete(entityId.getHBaseRowKey())
            .deleteColumns(hbaseColumnName.getFamilyBytes(), hbaseColumnName.getQualifierBytes(), upToTimestamp);
        updateBuffer(delete);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteCell(EntityId entityId, String family, String qualifier) throws IOException {
        deleteCell(entityId, family, qualifier, HConstants.LATEST_TIMESTAMP);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void deleteCell(EntityId entityId, String family, String qualifier, long timestamp)
        throws IOException {
        final ColumnName hbaseColumnName = ColumnName.create(family, qualifier);
        final Delete delete = new Delete(entityId.getHBaseRowKey())
            .deleteColumn(hbaseColumnName.getFamilyBytes(), hbaseColumnName.getQualifierBytes(), timestamp);
        updateBuffer(delete);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void setBufferSize(long bufferSize) throws IOException {
        synchronized (mInternalLock) {
            Preconditions.checkState(mState == State.OPEN,
                "Cannot set buffer size of BufferedWriter instance %s in state %s.", this, mState);
            Preconditions.checkArgument(bufferSize > 0,
                "Buffer size cannot be negative, got %s.", bufferSize);
            mMaxWriteBufferSize = bufferSize;
            if (mCurrentWriteBufferSize > mMaxWriteBufferSize) {
                flush();
            }
            mHTable.setWriteBufferSize(bufferSize);
        }
    }
    //endregion

    /**
     * {@inheritDoc}
     */
    @Override
    public void flush() throws IOException {
        synchronized (mInternalLock) {
            Preconditions.checkState(mState == State.OPEN,
                "Cannot flush BufferedWriter instance %s in state %s.", this, mState);
            if (mDeleteBuffer.size() > 0) {
                mHTable.delete(mDeleteBuffer);
                mDeleteBuffer.clear();
            }
            if (mPutBuffer.size() > 0) {
                for (Put put : mPutBuffer.values()) {
                    mHTable.put(put);
                }
                mHTable.flushCommits();
                mPutBuffer.clear();
            }
            mCurrentWriteBufferSize = 0L;
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void close() throws IOException {
        synchronized (mInternalLock) {
            flush();
            Preconditions.checkState(mState == State.OPEN,
                "Cannot close BufferedWriter instance %s in state %s.", this, mState);
            mState = State.CLOSED;
            DebugResourceTracker.get().unregisterResource(this);
            mHTable.close();
            mTable.release();
        }
    }

    /**
     * Add a Put to the buffer and update the current buffer size.
     *
     * @param entityId  the EntityId of the row to put into.
     * @param family    the byte[] representation of the hbase family to write into.
     * @param qualifier the byte[] representation of the hbase qualifier to write into.
     * @param timestamp the timestamp at which to write the value.
     * @param value     the byte[] representation of the value to write.
     * @throws IOException in case of an error on flush.
     */
    private void updateBuffer(EntityId entityId, byte[] family, byte[] qualifier,
                              long timestamp, byte[] value) throws IOException {
        synchronized (mInternalLock) {
            Preconditions.checkState(mState == State.OPEN,
                "Cannot write to BufferedWriter instance in state %s.", mState);
            if (mPutBuffer.containsKey(entityId)) {
                mCurrentWriteBufferSize -= mPutBuffer.get(entityId).heapSize();
                mPutBuffer.get(entityId).add(family, qualifier,
                    timestamp, value);
                mCurrentWriteBufferSize += mPutBuffer.get(entityId).heapSize();
            } else {
                final Put put = new Put(entityId.getHBaseRowKey())
                    .add(family, qualifier, timestamp, value);
                mPutBuffer.put(entityId, put);
                mCurrentWriteBufferSize += put.heapSize();
            }
            if (mCurrentWriteBufferSize > mMaxWriteBufferSize) {
                flush();
            }
        }
    }
    //endregion

    //region private methods

    /**
     * Add a Delete to the buffer and update the current buffer size.
     *
     * @param d A delete to add to the buffer.
     * @throws IOException in case of an error on flush.
     */
    private void updateBuffer(Delete d) throws IOException {
        synchronized (mInternalLock) {
            mDeleteBuffer.add(d);
            long heapSize = mDeleteSize;
            heapSize += ClassSize.align(ClassSize.ARRAY + d.getRow().length);
            mCurrentWriteBufferSize += heapSize;
            if (mCurrentWriteBufferSize > mMaxWriteBufferSize) {
                flush();
            }
        }
    }

    /**
     * States of a buffered writer instance.
     */
    private static enum State {
        UNINITIALIZED,
        OPEN,
        CLOSED
    }

    //endregion


}
